package net.zhangyun.reactor;

import net.zhangyun.utils.MyConsole;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;

public class FluxMain {
    private static ExecutorService executorService = new ForkJoinPool();

    public static void main(String[] args) throws Exception {
        FluxSample();
        FluxSample2();
    }

    private static void FluxSample() throws Exception {
        Flux<Integer> numbers = Flux.range(1, 100);

        executorService.submit(() -> {
            numbers.map(number -> {
                if (number <= 2) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    throw new RuntimeException("Out of range");
                }
                return number;
            }).take(3).subscribe(
                    MyConsole::printSingle,
                    MyConsole::printError
            );
        });

        Thread.sleep(3000);
    }

    private static void FluxSample2() throws InterruptedException {
        Flux.interval(Duration.ofSeconds(1)).subscribe(MyConsole::printSingle);
        Thread.sleep(5500);
    }
}
